package com.gitee.xhs.saveAndModifyingPointWriter.operatorState.listState;

import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.state.api.functions.StateBootstrapFunction;

import static com.gitee.xhs.operatorState.listState.OperatorListStateConstant.OPERATOR_LIST_STATE_DESCRIPTOR;

/**
 * @author BigHead
 * @date 2022/7/11 14:13
 * @description OperatorListStateNewSavePointsFunction
 */
public class OperatorListStateNewSavePointsFunction extends StateBootstrapFunction<Transaction> {

    private ListState<Long> listState;

    @Override
    public void processElement(Transaction value, Context ctx) throws Exception {
        listState.add(value.getTimestamp());
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        listState = context.getOperatorStateStore().getListState(OPERATOR_LIST_STATE_DESCRIPTOR);
    }
}
